package com.azarias.dfs.namenode.server.service;


import com.azarias.dfs.namenode.rpc.model.*;
import com.azarias.dfs.namenode.rpc.service.NameNodeServiceGrpc.NameNodeService;
import com.azarias.dfs.namenode.server.component.DataNodeManager;
import com.azarias.dfs.namenode.server.component.FSNamesystem;
import io.grpc.stub.StreamObserver;

/**
 * @description: name node的 rpc服务接口
 * @author: Azarias_Jx
 * @date: 2021/12/31 11:18
 */

public class NameNodeServiceImpl implements NameNodeService {
    //返回状态
    public final static Integer SUCCESS = 1;
    public final static Integer FAILURE = 2;
    public static final Integer STATUS_SHUTDOWN = 3;
    /**
     * 负责管理元数据的核心组件
     */
    private FSNamesystem namesystem;
    /**
     * 负责管理集群中所有的datanode的组件
     */
    private DataNodeManager datanodeManager;
    /**
     * 判断是否在运行中
     */
    private boolean isRunning;

    public NameNodeServiceImpl(FSNamesystem namesystem, DataNodeManager datanodeManager) {
        this.namesystem = namesystem;
        this.datanodeManager = datanodeManager;
    }

    @Override
    public void register(RegisterRequest request, StreamObserver<RegisterResponse> responseObserver) {

        //注册
        Boolean register = datanodeManager.register(request.getIp(), request.getHostname());
        if (register) {
            //返回给datanode注册信息
            RegisterResponse response = RegisterResponse.newBuilder().setStatus(SUCCESS).build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }

    }

    @Override
    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {
        //注册
        Boolean heartbeat = datanodeManager.heartbeat(request.getIp(), request.getHostname());


        if (heartbeat) {
            //返回给datanode注册信息
            HeartbeatResponse response = HeartbeatResponse.newBuilder().setStatus(SUCCESS).build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        }
    }

    @Override
    public void mkdir(MkdirRequest request, StreamObserver<MkdirResponse> responseObserver)  {
        try {

            Boolean mkdir = this.namesystem.mkdir(request.getPath());
            System.out.println("创建目录：path" + request.getPath());
            MkdirResponse response =null;
            if (!isRunning) {
                response = MkdirResponse.newBuilder()
                        .setStatus(STATUS_SHUTDOWN)
                        .build();
            } else {
                this.namesystem.mkdir(request.getPath());

                response = MkdirResponse.newBuilder()
                        .setStatus(SUCCESS)
                        .build();
            }
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     *优雅关闭
     * @param request
     * @param responseObserver
     */
    @Override
    public void shutdown(ShutdownRequest request, StreamObserver<ShutdownResponse> responseObserver) {
        this.isRunning = false;
        System.out.println("调用优雅关闭");
        this.namesystem.flush();
    }
}
